package com.xiefengf.rabbitmq.producer.config;

import com.xiefengf.rabbitmq.producer.utils.StringUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.batch.BatchingStrategy;
import org.springframework.amqp.rabbit.batch.SimpleBatchingStrategy;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.BatchingRabbitTemplate;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.transaction.RabbitTransactionManager;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.SerializerMessageConverter;
import org.springframework.boot.autoconfigure.amqp.SimpleRabbitListenerContainerFactoryConfigurer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Scope;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.scheduling.concurrent.ConcurrentTaskScheduler;

import java.util.HashMap;
import java.util.Map;


/**
 * @author xiefengf
 * @version 1.0
 * @date 2024/1/11 23:28
 *
 * RabbitMQ基础配置
 * 1、声明交换机
 * 2、声明队列
 * 3、绑定交换机和队列
 * 4、初始化交换机和队列
 *
 * RabbitMQ高级设置
 * 5、过期时间TTL属性设置
 * 6、死信队列
 * 7、消息确认机制（ack）
 *
 */
@Configuration
@Slf4j
public class RabbitConfig {

    @Bean
    public RabbitListenerContainerFactory<?> rabbitListenerContainerFactory(ConnectionFactory connectionFactory){
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        // json转消息
        factory.setMessageConverter(new Jackson2JsonMessageConverter());
        // 手动确认（手动加上的）
        factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        // 每次只能获取一条，处理完成才能获取下一条（手动加上的）
        factory.setPrefetchCount(1);
        return factory;
    }

    /**
     * ------------------------------------消息确认机制------------------------------------
     */
    @Bean("RabbitTemplate")
    public RabbitTemplate createRabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate();
        rabbitTemplate.setConnectionFactory(connectionFactory);
        // 设置开启Mandatory,才能触发回调函数,无论消息推送结果怎么样都强制调用回调函数
        rabbitTemplate.setMandatory(true);
        // 开启事务 --- 这里有问题。开启了事务 发送方就会报错。暂未找到原因
//        rabbitTemplate.setChannelTransacted(true);

        // 确认消息送到交换机(Exchange)回调 --> publisher-confirm-type: correlated
        // 如果消息没有到 exchange,则 confirm 回调,ack=false; 如果消息到达exchange,则confirm回调,ack=true
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                if (ack){
                    log.info("消息发送交换机成功：correlationData：{},ack：{},cause：{}",correlationData,ack,cause);
                }else {
                    log.info("消息发送交换机失败：correlationData：{},ack：{},cause：{}",correlationData,ack,cause);
                }
            }
        });

        // 确认消息送到队列(Queue)回调 --> publisher-returns: true
        // 如果exchange到queue成功,则不回调return;如果exchange到queue失败,则回调return(需设置mandatory=true,否则不回回调,消息就丢了)
        rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() {
            @Override
            public void returnedMessage(ReturnedMessage returnedMessage) {
                    log.info("确认消息送到队列结果：");
                    log.info("发送消息：{}", returnedMessage.getMessage());
                    log.info("回应码：{}", returnedMessage.getReplyCode());
                    log.info("回应信息：{}", returnedMessage.getReplyText());
                    log.info("交换机：{}", returnedMessage.getExchange());
                    log.info("路由键：{}", returnedMessage.getRoutingKey());
            }
        });
        return rabbitTemplate;
    }



    /**
     * 启动服务不默认先创建队列的话,在启动消费者时候,消费者服务监听会找不到对应的队列而报错,启动失败
     * 创建初始化RabbitAdmin对象
     * @param connectionFactory
     * @return
     */
    @Bean
    public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) {
        RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
        // 只有设置为 true，spring 才会加载 RabbitAdmin 这个类
        rabbitAdmin.setAutoStartup(true);
        // 创建fanout交换机和对列
        rabbitAdmin.declareExchange(fanoutOrderExchange());
        rabbitAdmin.declareQueue(fanoutWeixinQueue());
        rabbitAdmin.declareQueue(fanoutSmsQueue());
        rabbitAdmin.declareQueue(fanoutEmailQueue());
        // 创建direct交换机和对列
        rabbitAdmin.declareExchange(directOrderExchange());
        rabbitAdmin.declareQueue(directWeixinQueue());
        rabbitAdmin.declareQueue(directSmsQueue());
        rabbitAdmin.declareQueue(directEmailQueue());
        // 创建topic交换机和对列
        rabbitAdmin.declareExchange(topicOrderExchange());
        rabbitAdmin.declareQueue(topicWeixinQueue());
        rabbitAdmin.declareQueue(topicSmsQueue());
        rabbitAdmin.declareQueue(topicEmailQueue());
        // 过期时间TTL队列
        rabbitAdmin.declareExchange(directTtlExchange());
        rabbitAdmin.declareQueue(directQueueTtlQueue());
        rabbitAdmin.declareQueue(directMsgTtlQueue());
        // 消息确认机制(事务+confirm)
        rabbitAdmin.declareExchange(directAckExchange());
        rabbitAdmin.declareQueue(directAckQueue());
        // 死信队列
        rabbitAdmin.declareExchange(directDlxExchange());
        rabbitAdmin.declareQueue(directDlxQueue());
        // 批量处理消息队列
        rabbitAdmin.declareExchange(directBatchExchange());
        rabbitAdmin.declareQueue(directBatchQueue());
        // 惰性队列
        rabbitAdmin.declareQueue(directQueueLazyQueue());
        // 优先级队列
        rabbitAdmin.declareQueue(directQueuePriorityQueue());
        // 流量控制
        rabbitAdmin.declareQueue(directMaxLengthQueue());
        // 返回rabbitAdmin
        return rabbitAdmin;
    }



    /**
     * ------------------------------------fanout模式------------------------------------
     *
     * 1、声明交换机
     * name: 交换机名称
     * durable:是否持久化,默认是false,持久化队列：会被存储在磁盘上，当消息代理重启时仍然存在，暂存队列：当前连接有效
     * autoDelete:是否自动删除，当没有生产者或者消费者使用此队列，该队列会自动删除。
     */
    /**
     *  2、声明队列
     *  name: 队列名称
     *  durable:是否持久化,默认是false,持久化队列：会被存储在磁盘上，当消息代理重启时仍然存在，暂存队列：当前连接有效
     *  autoDelete:是否自动删除，当没有生产者或者消费者使用此队列，该队列会自动删除。
     *  exclusive:默认也是false，只能被当前创建的连接使用，而且当连接关闭后队列即被删除。此参考优先级高于durable
     *  arguments：设置队列属性
     *  （一般设置一下队列的持久化就好,其余两个就是默认false）
     */
    /**
     * 3、绑定交换机和队列
     */
    @Bean
    public FanoutExchange fanoutOrderExchange(){
        return new FanoutExchange(StringUtils.FANOUT_EXCHANGE_NAME,true,false);
    }
    @Bean
    public Queue fanoutWeixinQueue(){
        return new Queue(StringUtils.FANOUT_QUEUE_NAME_1,true,false,false);
    }
    @Bean
    public Queue fanoutSmsQueue(){
        return new Queue(StringUtils.FANOUT_QUEUE_NAME_2,true,false,false);
    }
    @Bean
    public Queue fanoutEmailQueue(){
        return new Queue(StringUtils.FANOUT_QUEUE_NAME_3,true,false,false);
    }
    @Bean
    public Binding bindingFanout1() {
        return BindingBuilder.bind(fanoutWeixinQueue()).to(fanoutOrderExchange());
    }
    @Bean
    public Binding bindingFanout2() {
        return BindingBuilder.bind(fanoutSmsQueue()).to(fanoutOrderExchange());
    }
    @Bean
    public Binding bindingFanout3() {
        return BindingBuilder.bind(fanoutEmailQueue()).to(fanoutOrderExchange());
    }



    /**
     * ------------------------------------direct模式------------------------------------
     *
     * 1、声明交换机
     * name: 交换机名称
     * durable:是否持久化,默认是false,持久化队列：会被存储在磁盘上，当消息代理重启时仍然存在，暂存队列：当前连接有效
     * autoDelete:是否自动删除，当没有生产者或者消费者使用此队列，该队列会自动删除。
     */
    /**
     *  2、声明队列
     *  name: 队列名称
     *  durable:是否持久化,默认是false,持久化队列：会被存储在磁盘上，当消息代理重启时仍然存在，暂存队列：当前连接有效
     *  autoDelete:是否自动删除，当没有生产者或者消费者使用此队列，该队列会自动删除。
     *  exclusive:默认也是false，只能被当前创建的连接使用，而且当连接关闭后队列即被删除。此参考优先级高于durable
     *  arguments：设置队列属性
     *  （一般设置一下队列的持久化就好,其余两个就是默认false）
     */
    /**
     * 3、绑定交换机和队列
     */
    @Bean
    public DirectExchange directOrderExchange(){
        return new DirectExchange(StringUtils.DIRECT_EXCHANGE_NAME,true,false);
    }
    @Bean
    public Queue directWeixinQueue(){
        return new Queue(StringUtils.DIRECT_QUEUE_NAME_1,true,false,false);
    }
    @Bean
    public Queue directSmsQueue(){
        return new Queue(StringUtils.DIRECT_QUEUE_NAME_2,true,false,false);
    }
    @Bean
    public Queue directEmailQueue(){
        return new Queue(StringUtils.DIRECT_QUEUE_NAME_3,true,false,false);
    }
    @Bean
    public Binding bindingDirect1() {
        return BindingBuilder.bind(directWeixinQueue()).to(directOrderExchange()).with(StringUtils.DIRECT_ROUTE_KEY_1);
    }
    @Bean
    public Binding bindingDirect2() {
        return BindingBuilder.bind(directSmsQueue()).to(directOrderExchange()).with(StringUtils.DIRECT_ROUTE_KEY_2);
    }
    @Bean
    public Binding bindingDirect3() {
        return BindingBuilder.bind(directEmailQueue()).to(directOrderExchange()).with(StringUtils.DIRECT_ROUTE_KEY_3);
    }



    /**
     * ------------------------------------topic模式------------------------------------
     *
     * 1、声明交换机
     * name: 交换机名称
     * durable:是否持久化,默认是false,持久化队列：会被存储在磁盘上，当消息代理重启时仍然存在，暂存队列：当前连接有效
     * autoDelete:是否自动删除，当没有生产者或者消费者使用此队列，该队列会自动删除。
     */
    /**
     *  2、声明队列
     *  name: 队列名称
     *  durable:是否持久化,默认是false,持久化队列：会被存储在磁盘上，当消息代理重启时仍然存在，暂存队列：当前连接有效
     *  autoDelete:是否自动删除，当没有生产者或者消费者使用此队列，该队列会自动删除。
     *  exclusive:默认也是false，只能被当前创建的连接使用，而且当连接关闭后队列即被删除。此参考优先级高于durable
     *  arguments：设置队列属性
     *  （一般设置一下队列的持久化就好,其余两个就是默认false）
     */
    /**
     * 3、绑定交换机和队列
     */
    @Bean
    public TopicExchange topicOrderExchange(){
        return new TopicExchange(StringUtils.TOPIC_EXCHANGE_NAME,true,false);
    }
    @Bean
    public Queue topicWeixinQueue(){
        return new Queue(StringUtils.TOPIC_QUEUE_NAME_1,true,false,false);
    }
    @Bean
    public Queue topicSmsQueue(){
        return new Queue(StringUtils.TOPIC_QUEUE_NAME_2,true,false,false);
    }
    @Bean
    public Queue topicEmailQueue(){
        return new Queue(StringUtils.TOPIC_QUEUE_NAME_3,true,false,false);
    }
    @Bean
    public Binding bindingTopic1() {
        return BindingBuilder.bind(topicWeixinQueue()).to(topicOrderExchange()).with(StringUtils.TOPIC_ROUTE_KEY_1);
    }
    @Bean
    public Binding bindingTopic2() {
        return BindingBuilder.bind(topicSmsQueue()).to(topicOrderExchange()).with(StringUtils.TOPIC_ROUTE_KEY_2);
    }
    @Bean
    public Binding bindingTopic3() {
        return BindingBuilder.bind(topicEmailQueue()).to(topicOrderExchange()).with(StringUtils.TOPIC_ROUTE_KEY_3);
    }



    /**
     * ------------------------------------过期队列------------------------------------
     *
     * 1、声明交换机
     * name: 交换机名称
     * durable:是否持久化,默认是false,持久化队列：会被存储在磁盘上，当消息代理重启时仍然存在，暂存队列：当前连接有效
     * autoDelete:是否自动删除，当没有生产者或者消费者使用此队列，该队列会自动删除。
     */
    /**
     *  2、声明队列
     *  name: 队列名称
     *  durable:是否持久化,默认是false,持久化队列：会被存储在磁盘上，当消息代理重启时仍然存在，暂存队列：当前连接有效
     *  autoDelete:是否自动删除，当没有生产者或者消费者使用此队列，该队列会自动删除。
     *  exclusive:默认也是false，只能被当前创建的连接使用，而且当连接关闭后队列即被删除。此参考优先级高于durable
     *  arguments：设置队列属性
     *  （一般设置一下队列的持久化就好,其余两个就是默认false）
     */
    /**
     * 3、绑定交换机和队列
     */

    @Bean
    public DirectExchange directTtlExchange(){
        return new DirectExchange(StringUtils.DIRECT_TTL_EXCHANGE_NAME,true,false);
    }
    @Bean
    public Queue directQueueTtlQueue(){
        // 这是队列中的消息都是过期消息
        Map<String , Object> arguments = new HashMap<>();
        // 设置过期时间，单位是毫秒
        arguments.put("x-message-ttl" , 30000);
        // 设定当前队列中，允许存放的最大消息数目
        arguments.put("x-max-length", 2);
        // 声明死信交换机 需要设定消息过期后丢入指定的死信交换机
        arguments.put("x-dead-letter-exchange", StringUtils.DIRECT_DLX_EXCHANGE_NAME);
        // 声明死信路由键。只有Direct和Topic需要设置
        arguments.put("x-dead-letter-routing-key", StringUtils.DIRECT_DLX_KEY);
        return new Queue(StringUtils.DIRECT_TTL_QUEUE_NAME_1,true,false,false,arguments);
    }
    @Bean
    public Queue directMsgTtlQueue(){
        // 这是队列是正常队列，消息设置为过期消息
        Map<String , Object> arguments = new HashMap<>();
        // 声明死信交换机 需要设定消息过期后丢入指定的死信交换机
        arguments.put("x-dead-letter-exchange", StringUtils.DIRECT_DLX_EXCHANGE_NAME);
        // 声明死信路由键。只有Direct和Topic需要设置
        arguments.put("x-dead-letter-routing-key", StringUtils.DIRECT_DLX_KEY);
        return new Queue(StringUtils.DIRECT_TTL_QUEUE_NAME_2,true,false,false,arguments);
    }
    @Bean
    public Binding bindingTtlFanout() {
        return BindingBuilder.bind(directQueueTtlQueue()).to(directTtlExchange()).with(StringUtils.DIRECT_TTL_KEY_1);
    }
    @Bean
    public Binding bindingNotTtlFanout() {
        return BindingBuilder.bind(directMsgTtlQueue()).to(directTtlExchange()).with(StringUtils.DIRECT_TTL_KEY_2);
    }



    /**
     * ------------------------------------死信队列------------------------------------
     *
     * 1、声明交换机
     * name: 交换机名称
     * durable:是否持久化,默认是false,持久化队列：会被存储在磁盘上，当消息代理重启时仍然存在，暂存队列：当前连接有效
     * autoDelete:是否自动删除，当没有生产者或者消费者使用此队列，该队列会自动删除。
     */
    /**
     *  2、声明队列
     *  name: 队列名称
     *  durable:是否持久化,默认是false,持久化队列：会被存储在磁盘上，当消息代理重启时仍然存在，暂存队列：当前连接有效
     *  autoDelete:是否自动删除，当没有生产者或者消费者使用此队列，该队列会自动删除。
     *  exclusive:默认也是false，只能被当前创建的连接使用，而且当连接关闭后队列即被删除。此参考优先级高于durable
     *  arguments：设置队列属性
     *  （一般设置一下队列的持久化就好,其余两个就是默认false）
     */
    /**
     * 3、绑定交换机和队列
     */
    @Bean
    public DirectExchange directDlxExchange(){
        return new DirectExchange(StringUtils.DIRECT_DLX_EXCHANGE_NAME,true,false);
    }
    @Bean
    public Queue directDlxQueue(){
        return new Queue(StringUtils.DIRECT_DLX_QUEUE_NAME_1,true,false,false);
    }
    @Bean
    public Binding bindingDlxDirect() {
        return BindingBuilder.bind(directDlxQueue()).to(directDlxExchange()).with(StringUtils.DIRECT_DLX_KEY);
    }

    /**
     * ------------------------------------批量处理消息------------------------------------
     */
    @Bean
    public DirectExchange directBatchExchange(){
        return new DirectExchange(StringUtils.DIRECT_BATCH_EXCHANGE_NAME,true,false);
    }
    @Bean
    public Queue directBatchQueue(){
        return new Queue(StringUtils.DIRECT_BATCH_QUEUE_NAME,true,false,false);
    }
    @Bean
    public Binding bindingBatchDirect(){
        return BindingBuilder.bind(directBatchQueue()).to(directBatchExchange()).with(StringUtils.DIRECT_BATCH_KEY);
    }
    @Bean("BatchingRabbitTemplate")
    public BatchingRabbitTemplate batchRabbitTemplate(ConnectionFactory connectionFactory) {
        // 创建 BatchingStrategy 对象，代表批量策略
        int batchSize = 16384; // 超过收集的消息数量的最大条数。
        int bufferLimit = 33554432; // 每次批量发送消息的最大内存
        int timeout = 30000; // 超过收集的时间的最大等待时长，单位：毫秒
        BatchingStrategy batchingStrategy = new SimpleBatchingStrategy(batchSize, bufferLimit, timeout);

        // 创建 TaskScheduler 对象，用于实现超时发送的定时器
        TaskScheduler taskScheduler = new ConcurrentTaskScheduler();

        // 创建 BatchingRabbitTemplate 对象
        BatchingRabbitTemplate batchTemplate = new BatchingRabbitTemplate(batchingStrategy, taskScheduler);
        batchTemplate.setConnectionFactory(connectionFactory);
        return batchTemplate;
    }


    /**
     * ------------------------------------消息确认机制(事务+confirm)------------------------------------
     */
    @Bean
    public DirectExchange directAckExchange(){
        return new DirectExchange(StringUtils.DIRECT_ACK_EXCHANGE_NAME,true,false);
    }
    @Bean
    public Queue directAckQueue(){
        return new Queue(StringUtils.DIRECT_ACK_QUEUE_NAME_1,true,false,false);
    }
    @Bean
    public Binding bindingAckDirect() {
        return BindingBuilder.bind(directAckQueue()).to(directAckExchange()).with(StringUtils.DIRECT_ACK_KEY);
    }

    /**
     * 配置rabbitmq事务
     *
     * @param connectionFactory
     * @return
     */
//    @Bean("rabbitTransactionManager")
//    public RabbitTransactionManager rabbitTransactionManager(ConnectionFactory connectionFactory) {
//        return new RabbitTransactionManager(connectionFactory);
//    }


    /**
     * ------------------------------------惰性队列------------------------------------
     */
    @Bean
    public Queue directQueueLazyQueue(){
        // 这是队列中的消息都是过期消息
        Map<String , Object> arguments = new HashMap<>();
        // 设置惰性队列，
        arguments.put("x-queue-mode" , "lazy");
        return new Queue("direct.lazy.queue",true,false,false,arguments);
    }

    /**
     * ------------------------------------优先级队列------------------------------------
     */
    @Bean
    public Queue directQueuePriorityQueue() {
        Map<String, Object> arguments = new HashMap<>();
        // 设置队列的最大优先级，优先级的取值范围为0~255，一般设置最大值为10
        arguments.put("x-max-priority", 10);
        return new Queue("direct.priority.queue",true,false,false,arguments);
    }

    /**
     * ------------------------------------流量控制------------------------------------
     */
    @Bean
    public Queue directMaxLengthQueue() {
        Map<String, Object> arguments = new HashMap<>();
        // 设置队列的最大长度
        arguments.put("x-max-length", 10);
        // 设置队列的队列最大容量
        arguments.put("x-max-length-bytes", 1024);
        return new Queue("direct.maxlength.queue",true,false,false,arguments);
    }

}
